package com.zhaorujie.rpc.core.remoting.provider;

import com.zhaorujie.rpc.core.RpcProperties;
import com.zhaorujie.rpc.core.exception.RpcException;
import com.zhaorujie.rpc.core.register.Register;
import com.zhaorujie.rpc.core.register.RegisterFactory;
import com.zhaorujie.rpc.core.remoting.netty.server.NettyServer;
import com.zhaorujie.rpc.core.remoting.params.RpcRequest;
import com.zhaorujie.rpc.core.remoting.params.RpcResponse;
import com.zhaorujie.rpc.core.serialize.RpcSerialize;
import com.zhaorujie.rpc.core.serialize.SerializeFactory;
import com.zhaorujie.rpc.core.utils.IpUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author 赵汝杰
 * @version v1.0
 * @description RpcProvideFactroy
 * @date 2024/3/5
 **/
public class RpcProviderFactory {

    private static final Logger logger = LoggerFactory.getLogger(RpcProviderFactory.class);

    /**
     * rpc远程调用的端口
     */
    private int port;

    /**
     * 当前项目启动的ip
     */
    private String ip;

    /**
     * 核心线程池数
     */
    private int corePoolSize;

    /**
     * 最大线程池数
     */
    private int maxPoolSize;

    /**
     * 注册中心地址
     */
    private String registryAddr;

    /**
     * PRC配置文件
     */
    private RpcProperties rpcProperties;

    /**
     * RPC序列化实例
     */
    private RpcSerialize serializer;

    /**
     * 注册中心实例
     */
    private Register register;

    /**
     * netty服务端实例
     */
    private NettyServer nettyServer;

    /**
     * 服务地址：服务提供者的 ip:port
     */
    private String serviceAddr;

    private final Map<String, Map<String, Object>> serviceData = new ConcurrentHashMap<>();

    /**
     * 初始化配置
     * @param rpcProperties 配置文件信息
     */
    public void init(RpcProperties rpcProperties) {
        this.rpcProperties = rpcProperties;
        this.port = rpcProperties.getPort();
        this.registryAddr = rpcProperties.getRegisterAddr();
        // 获取序列化实例对象
        this.serializer = SerializeFactory.getInstance(rpcProperties.getSerializeType());
        // 获取注册中心实例对象
        this.register = RegisterFactory.getInstance(rpcProperties.getRegistryType());

        this.corePoolSize = rpcProperties.getCorePoolSize();
        this.maxPoolSize = rpcProperties.getMaxPoolSize();

        // 创建netty服务端实例
        this.nettyServer = new NettyServer(this);
        // 注册服务关闭钩子
        nettyServer.registerShutdownHook();
        // 开启netty服务端
        nettyServer.start();
        // 获取服务启动的注册地址
        this.serviceAddr = IpUtil.getIpPort(IpUtil.getIp(), rpcProperties.getPort());
        // 初始化注册中心
        register.init(registryAddr, rpcProperties.getEnv());
        // 把接口信息注册到注册中心上
        register.register(serviceData.keySet(), serviceAddr);
    }

    /**
     * 构建服务的key
     * @param interfaceName 接口全路径名称
     * @param version 版本号
     * @return
     */
    public static String buildServiceKey(String interfaceName, String version) {
        if (Objects.nonNull(version) && !version.trim().isEmpty()) {
            interfaceName += "#" + version;
        }
        return interfaceName;
    }

    /**
     * 添加服务可以和bean对象的关系
     *
     * @param interfaceName
     * @param version
     * @param timeout
     * @param serviceBean
     */
    public void addService(String interfaceName, String version, long timeout, Object serviceBean) {
        String serviceKey = buildServiceKey(interfaceName, version);
        Map<String, Object> map = new HashMap<>();
        map.put("timeout", timeout);
        map.put("serviceBean", serviceBean);
        serviceData.put(serviceKey, map);
        logger.info("rpc, provider factory add service success. serviceKey = {}, serviceBean = {}", serviceKey, serviceBean.getClass());
    }

    /**
     * 调用服务
     * @param rpcRequest 请求
     * @return 响应结果
     */
    public RpcResponse invokeService(RpcRequest rpcRequest) {
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setRequestId(rpcRequest.getRequestId());

        // 获取服务调用的bean对象
        String serviceKey = buildServiceKey(rpcRequest.getClassName(), rpcRequest.getVersion());
        Map<String, Object> map = serviceData.get(serviceKey);
        Object serviceBean = map.get("serviceBean");
        if (Objects.isNull(serviceBean)) {
            rpcResponse.setThrowable(new RpcException(String.format("The serviceKey[%s] not found.", serviceKey)));
            return rpcResponse;
        }
        // 获取服务的超时时间
        long timeout = (long) map.get("timeout");
        if (timeout != -1 && (System.currentTimeMillis() - rpcRequest.getCreateTime() > timeout)) {
            rpcResponse.setThrowable(new RpcException("service provider interface timeout."));
            return rpcResponse;
        }

        // 通过CompletableFuture的方式为请求设置超时时间
        CompletableFuture<Boolean> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Class<?> serviceClass = serviceBean.getClass();
                String methodName = rpcRequest.getMethodName();
                Class<?>[] parameterTypes = rpcRequest.getParameterTypes();
                Object[] parameters = rpcRequest.getParameters();

                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceBean, parameters);
                rpcResponse.setResult(result);
            } catch (Exception e) {
                logger.error("rpc provider invokeService error.", e);
                rpcResponse.setThrowable(e.getCause());
            }
            return true;
        }).completeOnTimeout(false, timeout, TimeUnit.MILLISECONDS);

        Boolean result = completableFuture.join();
        if (!result) {
            rpcResponse.setThrowable(new RpcException("interface call timeout"));
        }

        return rpcResponse;
    }

    /**
     * 服务销毁
     */
    protected void destroy() {
        if (!serviceData.isEmpty()) {
            register.removeRegister(serviceData.keySet(), serviceAddr);
        }
        register.destroy();
    }


    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public int getCorePoolSize() {
        return corePoolSize;
    }

    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    public int getMaxPoolSize() {
        return maxPoolSize;
    }

    public void setMaxPoolSize(int maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }

    public String getRegistryAddr() {
        return registryAddr;
    }

    public void setRegistryAddr(String registryAddr) {
        this.registryAddr = registryAddr;
    }

    public RpcProperties getRpcProperties() {
        return rpcProperties;
    }

    public void setRpcProperties(RpcProperties rpcProperties) {
        this.rpcProperties = rpcProperties;
    }

    public RpcSerialize getSerializer() {
        return serializer;
    }

    public void setSerializer(RpcSerialize serializer) {
        this.serializer = serializer;
    }

    public NettyServer getNettyServer() {
        return nettyServer;
    }

    public void setNettyServer(NettyServer nettyServer) {
        this.nettyServer = nettyServer;
    }



}
